#include "log.h"
#include <sys/epoll.h>
#include <sys/types.h>
#include <netinet/tcp.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <signal.h>
#include "rtxpSession.h"
#include <memory>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include "ioevent.h"
#include <thread>
#include <semaphore.h>ca

#define UNUSE(var) ((void)var)

typedef struct _WorkEvent
{
	char event_type; // 0: read event.1:write event 2: close event
	int fd;
	std::chrono::time_point<std::chrono::steady_clock> version;
} WorkEvent;

typedef struct _ThreadStatus
{
	int status; // 1:busy or idle
	RingBuffer<WorkEvent *> work_list;
	sem_t worker_sem;
	std::thread tid;

} ThreadStatus;

ThreadStatus **thread_status;
int io_epfd;

static void add_task_queue(int thread_index, int fd, const std::chrono::time_point<std::chrono::steady_clock> &version, int type)
{
	WorkEvent *event = new WorkEvent();
	event->fd = fd;
	event->event_type = type;
	event->version = version;
	thread_status[thread_index]->work_list.push_task(event);
	sem_post(&(thread_status[thread_index]->worker_sem)); // 激活worker thread

	return;
}


static void close_session(int fd)
{
	if (fd < 0)
		return;
	if (events[fd] == NULL || events[fd]->session_handler == NULL)
		return;
	if (GET(status) == INACTIVATE)
		return;

	GET(status) = INACTIVATE;

	if (events[fd]->protocol == 1)
	{
		int video_rtp_fd = RTSP(tracks[VIDEOTRACK].rtp_fd); // 关闭video rtp session
		if (video_rtp_fd != -1)
		{
			delete (events[video_rtp_fd]->session_handler);
			events[video_rtp_fd]->session_handler = nullptr;
			close(video_rtp_fd);
		}

		int audio_rtp_fd = RTSP(tracks[AUDIOTRACK].rtp_fd); // 关闭audio rtp session
		if (audio_rtp_fd != -1)
		{
			delete (events[audio_rtp_fd]->session_handler);
			events[audio_rtp_fd]->session_handler = nullptr;
			close(audio_rtp_fd);
		}

		int video_rtcp_fd = RTSP(tracks[VIDEOTRACK].rtcp_fd); // 关闭video rtcp session
		if (video_rtcp_fd != -1)
		{

			delete (events[video_rtcp_fd]->session_handler);
			events[video_rtcp_fd]->session_handler = nullptr;
			close(video_rtcp_fd);
		}

		int audio_rtcp_fd = RTSP(tracks[AUDIOTRACK].rtcp_fd); // 关闭audio rtcp session
		if (audio_rtcp_fd != -1)
		{
			delete (events[audio_rtcp_fd]->session_handler);
			events[audio_rtcp_fd]->session_handler = nullptr;
			close(audio_rtcp_fd);
		}
		delete (events[fd]->session_handler); // 关闭rtsp session
		events[fd]->session_handler = nullptr;
		close(fd);
	}
}

#define del_event(fd)                                                                                        \
	{                                                                                                        \
		LOG("############################del_event:%d\n", fd);                                               \
		if (fd < 0 || events[fd] == NULL)                                                                    \
		{                                                                                                    \
			LOG("args invalid:%d\n", fd);                                                                    \
			return;                                                                                          \
		}                                                                                                    \
		if (events[fd]->protocol == 2 || (events[fd]->protocol == 0 && events[fd]->session_handler != NULL)) \
		{                                                                                                    \
			epoll_ctl(io_epfd, EPOLL_CTL_DEL, RTP(read_timer), NULL);                                        \
		}                                                                                                    \
		else if (events[fd]->protocol == 1)                                                                  \
			close_session(GET(sock_fd));                                                                     \
	}

static inline int do_read_event(int fd)
{
	if (events[fd] == NULL)
	{
		LOG("session[%d] is null\n", fd);
		return 0;
	}
	if (events[fd]->session_handler == NULL)
	{
		LOG("session[%d]->session_handler is null\n", fd);
		return 0;
	}

	if (GET(status) == ACTIVATE)
	{
		if (events[fd]->protocol == 1) // rtsp事件
		{
			return RTSP(read_rtsp_data()); // rtsp信令
		}
		else if (events[fd]->protocol == 0) // 定时器事件，读取视频文件
		{
			uint64_t t;   // unuse
			read(fd, &t, sizeof(uint64_t));
			UNUSE(t);
			return RTP(read_file_data(fd));
		}
		else if (events[fd]->protocol == 2) // rtp事件
		{
			char buff[4096] = "";  
			recv(fd, buff, sizeof(buff), 0);
			UNUSE(buff); // 不关心, 正常也不会走这个逻辑
			return 0;
		}
		else if (events[fd]->protocol == 3) // rtcp事件
		{
			char buff[4096] = "";
			struct sockaddr_in si_other;
			int slen = sizeof(si_other);
			recvfrom(fd, buff, sizeof(buff), 0, (struct sockaddr *)&si_other, (socklen_t *)&slen);
			UNUSE(buff);   // 不关心rtcp，收到rtcp就返回sdes
			RTCP(send_rtcp());
			return 0;
		}
	}
	LOG("do read event return -1:%d--[%d]\n", fd, GET(status));
	return -1;
}

void modify_event(int epfd, int fd, uint32_t events, int op)
{
	struct epoll_event ev;
	memset(&ev, 0x00, sizeof(ev));
	ev.events = events;
	ev.data.fd = fd;
	epoll_ctl(epfd, op, fd, &ev);
}

static inline int do_write_event(int fd)
{
	int ret = -1;
	if (events[fd] == NULL || events[fd]->session_handler == NULL)
	{
		LOG("send_data error:socket [%d] is not connectd\n", fd);
		return 0;
	}

	while (1)
	{
		int n = 0;
		shared_ptr<WriteData> write_data;
		n = GET(write_buffer).get_task(write_data);
		if (n < 0)
		{
			ret = 0;
			break;
		}

		if (events[fd]->protocol == 1) // tcp event
		{
			n = send(fd, write_data->data + GET(write_buffer).get_nonread(), write_data->len - GET(write_buffer).get_nonread(), MSG_NOSIGNAL);

			LOG("write data:[%d]--%s\n", n, write_data->data);
		}
		else if (events[fd]->protocol == 2 || events[fd]->protocol == 3) // rtp/rtcp事件
		{
			n = sendto(fd, write_data->data + GET(write_buffer).get_nonread(), write_data->len - GET(write_buffer).get_nonread(), 0,
					   (struct sockaddr *)&GET(remote_addr), sizeof(struct sockaddr_in));
		}

		if (n <= 0)
		{
			if (n < 0 && errno == EAGAIN)
			{ // socket写缓冲区满
				ret = 0;
				modify_event(io_epfd, fd, EPOLLET | EPOLLERR | EPOLLIN, EPOLL_CTL_MOD);
				break;
			}
			LOG("sendto failed:%s\n", strerror(errno));
			break;
		}

		GET(write_buffer).set_nonread(write_data->len - n);

		if (GET(write_buffer).get_nonread() == 0) // 队列中一个节点的所有数据都写完
		{
			modify_event(io_epfd, fd, EPOLLET | EPOLLERR | EPOLLIN, EPOLL_CTL_MOD);
			GET(write_buffer).done_task();
		}
	}
	return ret;
}

/** 每一个worker线程可以处理n个连接
 *  但每一个连接要由一个线程处理*/
void worker_thread(int index)
{
	thread_status[index]->status = 0;
	sem_init(&(thread_status[index]->worker_sem), 0, 0);
	while (1)
	{
		sem_wait(&(thread_status[index]->worker_sem));

		thread_status[index]->status = 1; // busy
		WorkEvent *worker_event;
		if (thread_status[index]->work_list.get_task2(worker_event) < 0)
		{
			LOG("worker_event is nullptr\n");
			continue;
		}
		int fd = worker_event->fd;

		if (events[fd] == NULL || events[fd]->session_handler == NULL)
			continue;

		if (GET(thread_index) != index)
		{
			LOG("thread_index not match\n");
			continue;
		}

		if (GET(version) != worker_event->version)
		{
			LOG("version not match\n");
			continue;
		}

		if (worker_event->event_type == 0) // 可读事件
		{
			if (do_read_event(fd) < 0)
				del_event(fd);
		}
		else if (worker_event->event_type == 1) // 可写事件
		{
			if (do_write_event(fd) < 0)
				del_event(fd);
		}
		else if (worker_event->event_type == 2) // 关闭事件
		{
			del_event(fd);
		}
		delete worker_event;
		thread_status[index]->status = 0;
		continue;
	}

	return ;
}

static int get_idle_thread(int thread_num)
{
	int j = 0;
	for (; j < thread_num; j++)
	{

		if (thread_status[j]->status == 0)
		{ // idle

			return j;
		}
	}

	return -1;
}


static int create_worker_threads(int num)
{
	thread_status = new ThreadStatus *[num];
	for (int i = 0; i < num; i++)
	{
		thread_status[i] = new ThreadStatus();
		thread_status[i]->status = 0; // 1:busy or idle
		thread_status[i]->tid = thread(worker_thread, i);
		thread_status[i]->tid.detach();
	}

	return 0;
}

void schedule_thread(int thread_num)
{

	if (create_worker_threads(thread_num) < 0)
		return ;

	struct epoll_event io_event_set[MAXCONN];
	io_epfd = epoll_create(1024);
	while (1)
	{
		int fd_count = epoll_wait(io_epfd, io_event_set, sizeof(io_event_set) / sizeof(struct epoll_event), -1);
		if (fd_count < 0)
		{
			LOG("epoll_wait failed:%s\n", strerror(errno));
			continue;
		}

		int i = 0;
		for (; i < fd_count; i++)
		{
			int fd = io_event_set[i].data.fd;
			if (events[fd] == NULL)
			{
				LOG("events[%d] is null\n", fd);
				continue;
			}

			if (GET(thread_index) == -1)
			{
				int id = get_idle_thread(thread_num);
				if (id > 0)
					GET(thread_index) = id;
				else
					GET(thread_index) = (fd % thread_num); // 任务均匀分配到每一个线程.
			}

			if (io_event_set[i].events & EPOLLIN)
			{
				add_task_queue(GET(thread_index), fd, GET(version), 0);
			}
			else if (io_event_set[i].events & EPOLLOUT)
			{
				add_task_queue(GET(thread_index), fd, GET(version), 1);
			}
			else if (io_event_set[i].events & EPOLLERR)
			{
				LOG("socket get EPOLLERR:%d\n", io_event_set[i].data.fd);
				del_event(io_event_set[i].data.fd);
			}
		}
	}

	return ;
}

void create_schedule_thread(int num)
{
	std::thread tid = std::thread(schedule_thread, num);
	tid.detach();
}
